package com.example.source;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
import org.apache.flink.connector.datagen.source.DataGeneratorSource;
import org.apache.flink.connector.datagen.source.GeneratorFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Created with IntelliJ IDEA.
 * ClassName: DataSource
 * Package: com.example.source
 * Description:
 * User: fzykd
 *
 * @Author: LQH
 * Date: 2023-07-18
 * Time: 15:37
 */

//数据自动生成器 读取数据
public class DataSource {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //数据生成器参数
        //1.实现GeneratorFunction接口 重写map方法 输入类型固定是Long
        //2.第二个参数是Long类型 自动生成的数字类型序列(从1自增) 的大小范围
        //3.限速的策略 比如每秒生成几条数据
        //4.返回的类型
        DataGeneratorSource<String> dataSource = new DataGeneratorSource<>(
                new GeneratorFunction<Long, String>() {
                    @Override
                    public String map(Long aLong) throws Exception {
                        return "Number: " + aLong;
                    }
                },
                //自动生成一个Long的最大值实现一个无界流的效果
                //设置了多个并行度 最大值按范围进行划分
                Long.MAX_VALUE,
                RateLimiterStrategy.perSecond(10),
                Types.STRING
        );
        env.fromSource(dataSource, WatermarkStrategy.noWatermarks(),"data").print();
        env.execute();
    }

}
